# 服务ip
# 回归的偏移量
# 是否支持第二参数为时间，第三个有值表示第二个参数为时间，否则为offset
KAFKA_HOME=/opt/soft/client/kafka
DATA_HANDLER_CONSUMER="uip_engrgy_device-data_group1"
TOPIC="device-data"
KAFKA_SERVER=$1
OFFSET_NUM=$2
function relaOffset(){

  res=$(${KAFKA_HOME}/bin/kafka-console-consumer.sh --bootstrap-server ${KAFKA_SERVER}:9092 --topic ${TOPIC} --partition 0 --offset $1 --max-messages 1 --property print.headers=true | awk '{print $1}' | sed -n 's/.*tb_msg_md_ts:\(.*\)/\1/gp')
  if [[ ${res##*:} -ge $3 ]]
  then
    echo 111111111
  fi
  echo $res
}
# ${KAFKA_HOME}/bin/kafka-consumer-groups.sh --bootstrap-server ${KAFKA_SERVER}:9092 --list
client_info=$(${KAFKA_HOME}/bin/kafka-consumer-groups.sh --bootstrap-server ${KAFKA_SERVER}:9092 --describe --group ${DATA_HANDLER_CONSUMER} | sed -n '3p' | awk '{print $(NF-1)}')
if [[ "$client_info" =~ ^\/.* ]]
then
  echo "found clients , please stop it first, I will do nothing now ....   "${client_info}
  exit
fi

# select start offset
START_OFFSET=$(${KAFKA_HOME}/bin/kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list ${KAFKA_SERVER}:9092 --topic ${TOPIC} --time -2)
END_OFFSET=$(${KAFKA_HOME}/bin/kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list ${KAFKA_SERVER}:9092 --topic ${TOPIC} --time -1)
if [[ -n $3 ]]
then
  relaOffset ${START_OFFSET##*:} ${END_OFFSET##*:} 1718167762100
fi
echo "${START_OFFSET##*:}--->${END_OFFSET##*:}"
if [[ -n $2 ]]
then
  if [[ -n $2 && ${START_OFFSET##*:} -le ${OFFSET_NUM} ]]
  then
    res=$(${KAFKA_HOME}/bin/kafka-console-consumer.sh --bootstrap-server ${KAFKA_SERVER}:9092 --topic ${TOPIC} --partition 0 --offset ${OFFSET_NUM} --max-messages 1 --property print.headers=true)
    echo $res
    ${KAFKA_HOME}/bin/kafka-consumer-groups.sh --bootstrap-server ${KAFKA_SERVER}:9092 --topic ${TOPIC} --group ${DATA_HANDLER_CONSUMER} --reset-offsets --to-offset ${OFFSET_NUM} --execute
  else
    echo "sorry , the offset ${OFFSET_NUM} data have clear"
  fi
fi
